package cn.itcast.flink.allowlateness;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * Author itcast
 * Date 2021/12/2 16:32
 * Desc 统计窗口内每10s 事件时间的统计，统计窗口内出现的次数并将窗口的开始时间和结束时间打印到控制台
 */
public class AllowLatenessDemo {
    public static void main(String[] args) throws Exception {
        //获取流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //获取socket数据源
        DataStreamSource<String> source = env.socketTextStream("node1", 9999);
        //hello,1636169401
        //将每行数据转换成 Tuple2<String,Long>
        SingleOutputStreamOperator<Tuple2<String, Long>> mapStream = source.map(new MapFunction<String, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(String str) throws Exception {
                String[] arrs = str.split(",");
                return Tuple2.of(
                        arrs[0],
                        Long.parseLong(arrs[1])
                );
            }
        });
        //分配水位线，最大延迟3s
        SingleOutputStreamOperator<Tuple2<String, Long>> mapWithWatermarkStream = mapStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<String, Long>>() {
                                    @Override
                                    public long extractTimestamp(Tuple2<String, Long> element, long recordTimestamp) {
                                        return element.f1 * 1000;
                                    }
                                })
                );
        //根据单词进行分组
        SingleOutputStreamOperator<Tuple2<String, Integer>> result = mapWithWatermarkStream
                .keyBy(new KeySelector<Tuple2<String, Long>, String>() {
            @Override
            public String getKey(Tuple2<String, Long> value) throws Exception {
                return value.f0;
            }
        })
                //滚动事件时间窗口为5s
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                //允许最大严重乱序时间为2s
                .allowedLateness(Time.seconds(2))
             //process处理，对窗口数据中的元素进行统计，生成[单词,出现次数]，并将窗口开始时间和结束时间打印到控制台
                .process(new ProcessWindowFunction<Tuple2<String, Long>, Tuple2<String, Integer>, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<Tuple2<String, Long>> elements, Collector<Tuple2<String, Integer>> out) throws Exception {
                        int size = 0;
                        for (Tuple2<String, Long> element : elements) {
                            size++;
                        }
                        out.collect(Tuple2.of(s, size));
                        System.out.println("当前窗口开始时间："+context.window().getStart()+" 窗口结束时间："+context.window().getEnd());
                    }
                });
        //打印结果
        result.print();
        //执行流环境
        env.execute();
    }
}